﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Fetches.Fetch.Dto;
using Fetches.Fetch.Factory.Persistence;
using Fetches.Fetch.Fetcher;
using Fetches.Fetch.Fetcher.AntiSpider;
using Fetches.Fetch.Fetcher.LinkExtractor;
using Fetches.Fetch.PagePool;
using Fetches.Fetch.PagePool.Consume;
using Fetches.Fetch.Proxy;
using Fetches.Fetch.Scheduler;
using Fetches.Fetch.Utils;
using Fetches.Fetch.Utils.FileHosting;
using Fetches.Fetch.Utils.KvStore;
using LightInject;
using Serilog;

namespace Fetches.Fetch.Factory
{
    /// <summary>
    /// 一个LocalFetchFactory对应一个FetchFactoryUnit
    /// </summary>
    public class LocalFetchFactory : AbstractFetchFactory
    {
        #region Field

        /// <summary>
        /// FactoryId
        /// </summary>
        public readonly long Id;
        /// <summary>
        /// Factory内部调度信息
        /// </summary>
        public readonly UnitSchedules UnitSchedules;
        

        #endregion

        public static FactoryConfig CurrentFactoryConfig { get; protected set; }
        private static IServiceContainer _serviceContainer;

        protected readonly IRuntimeDataService RuntimeDataService;
        private readonly ConcurrentDictionary<long, FetchTask> _fetcher2TaskMap = new ConcurrentDictionary<long, FetchTask>();
        

        #region 组件缓存

        private readonly ConcurrentDictionary<long, AbstractFetcher> _fetchers =
            new ConcurrentDictionary<long, AbstractFetcher>();

        private readonly ConcurrentDictionary<long, AbstractPagePool> _pagePools =
            new ConcurrentDictionary<long, AbstractPagePool>();

        private readonly ConcurrentDictionary<long, IProxyPool> _proxyPools =
            new ConcurrentDictionary<long, IProxyPool>();

        private readonly ConcurrentDictionary<long, ILinkExtractor> _linkExtractors =
            new ConcurrentDictionary<long, ILinkExtractor>();

        private readonly ComponentBox<AntiSpiderDetector> _detectors;
        private readonly ComponentBox<AbstractScheduler> _schedulers;
        private readonly FileHost FileHost;
        #endregion


        #region Factory创建及关闭

        public static LocalFetchFactory CreateWithFilePersistence(string filePath = null)
        {
            return new LocalFetchFactory(new StandaloneRuntimeDataService(filePath), "");
        }

        static LocalFetchFactory()
        {
            //解决部分网页gbk乱码问题
            Encoding.RegisterProvider(CodePagesEncodingProvider.Instance);

            RegisterComponents();
        }

        public LocalFetchFactory(IRuntimeDataService runtimeDataService, string endpoint = "")
        {
            RuntimeDataService = runtimeDataService;

            try
            {
                //读取配置，写入CurrentFactoryConfig用于Web等第三方读取
                CurrentFactoryConfig = runtimeDataService.ReadFactoryConfig();
            }
            catch (Exception e)
            {
                Log.Fatal("Factory startup failed with {exception}", e);
                throw new Exception("无法读取RuntimeData，系统无法正常启动");
            }

            Id = IdHelper.FactoryUnitIdService.GenerateId();

            var kvStore = GetInstance<IKvStore>(CurrentFactoryConfig.HostFileKvStoreType);
            kvStore.Initialize(CurrentFactoryConfig.HostFileKvStoreConfig, 0);
            var hostFileProvider = GetInstance<IHostFileProvider>(CurrentFactoryConfig.HostFileProviderType);
            hostFileProvider.Initialize(CurrentFactoryConfig.HostFileProviderConfig);
            FileHost = new FileHost(hostFileProvider, kvStore);

            _detectors = new ComponentBox<AntiSpiderDetector>(
                s => new AntiSpiderDetector(),
                (detector, taskId) =>
                {

                });

            _schedulers=new ComponentBox<AbstractScheduler>(
                GetInstance<AbstractScheduler>,
                (scheduler, taskId) =>
                {
                    scheduler.Initialize(CurrentFactoryConfig.SchedulerConfig, taskId);
                    //由Scheduler来通知爬取结束，中间的进度通知的话由Factory来统计
                    scheduler.FinishNotify = () => FetchFinished(taskId);
                }
                );

            var supportTypes = from t in typeof(IFetchFactory).Assembly.GetTypes()
                               where TypeHelper.IsSubClassOf(t, typeof(AbstractFetcher)) && !t.IsAbstract
                               select t.FullName;


            //登记调度信息
            UnitSchedules = new UnitSchedules(Id)
            {
                SupportedFetcherTypes = supportTypes,
                Endpoint = endpoint
            };
            UpdateScheduleInfo();
            Log.Information("FetchFactory {id} with config {@config}is ready for work,endpoint is {endpoint}",
                UnitSchedules.Id, CurrentFactoryConfig, endpoint);
        }

        /// <summary>
        ///     关闭Factory，释放资源
        /// </summary>
        public void Shutdown()
        {
            StopFetchers(_fetchers.Keys.ToList());
            _schedulers.ReleaseAll();
            FileHost.Dispose();
            Log.Information("FetchFactory {id} on endpoint {endpoint} is shutting down ",
                UnitSchedules.Id, UnitSchedules.Endpoint);
        }

        #endregion

        #region 平台组件获取
        /// <summary>
        /// 注册程序集
        /// </summary>
        private static void RegisterComponents()
        {
            //注册程序集,之后可以通过类的全名来获得实例
            _serviceContainer = new ServiceContainer();
            _serviceContainer.RegisterAssembly(typeof(IFetchFactory).Assembly);
        }

        /// <summary>
        /// 消费本地PagePool时使用LightInject实例化时直接无任何提示退出，所以可能换autofac
        /// 不过实测发现不是lightInject的锅，甚至直接new OnDiskPagePool也直接退出
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="typeName"></param>
        /// <returns></returns>
        private T GetInstance<T>(string typeName) where T : class
        {
            return _serviceContainer.GetInstance<T>(typeName);
        }


        /// <summary>
        /// </summary>
        /// <param name="uniqueId">唯一ID会作为回执的ID返回</param>
        /// <returns></returns>
        public async Task<AbstractPagePool> GetPagePool(long uniqueId)
        {
            return await Task.Run(() =>
            {
                lock (_pagePools)
                {
                    if (_pagePools.TryGetValue(uniqueId, out var instance)) return instance;

                    instance = GetInstance<AbstractPagePool>(CurrentFactoryConfig.PagePoolType);
                    instance.Initialize(CurrentFactoryConfig.PagePoolConfig, uniqueId);
                    _pagePools.TryAdd(uniqueId, instance);
                    return instance;
                }

            });

        }


        public async Task<IProxyPool> GetProxyPool(long uniqueId)
        {
            return await Task.Run(() =>
            {
                lock (_proxyPools)
                {
                    if (_proxyPools.TryGetValue(uniqueId, out var instance)) return instance;

                    instance = GetInstance<IProxyPool>(CurrentFactoryConfig.ProxyPoolType);
                    instance.Initialize(CurrentFactoryConfig.ProxyPoolConfig, uniqueId);
                    _proxyPools.TryAdd(uniqueId, instance);
                    return instance;
                }

            });

        }

        public async Task<IProcessCoordinator> GetProcessCoordinator(long uniqueId)
        {
            return await Task.Run(() =>
            {
                var instance = GetInstance<IProcessCoordinator>(CurrentFactoryConfig.ProcessCoordinatorType);
                instance.Initialize(CurrentFactoryConfig.ProcessCoordinatorConfig, uniqueId);
                return instance;
            });

        }

        #endregion

        #region 任务组件获取

        public ILinkExtractor GetLinkExtractor(long uniqueId, FetchOrder order)
        {
            lock (_linkExtractors)
            {
                if (_linkExtractors.TryGetValue(uniqueId, out var instance)) return instance;

                //不是默认Extractor的情况下会有点复杂
                if (string.IsNullOrEmpty(order.LinkExtractor))
                {
                    instance = new PatternMatchExtractor()
                        .AddIndexPatterns(order.IndexPatterns)
                        .AddContentPatterns(order.ContentPatterns);
                }
                else
                {
                    instance = GetInstance<ILinkExtractor>(order.LinkExtractor);
                    instance.Initialize(order.LinkExtractorRule);
                }
                _linkExtractors.TryAdd(uniqueId, instance);
                return instance;
            }

        }


        #endregion

        #region 数据采集

        public FetchReceipt TakeFetchOrder(FetchOrder order)
        {
            return ActuallyFetch(order);
        }

        protected override FetchReceipt ActuallyFetch(FetchOrder order)
        {
            var task = FetchOrder2Task(order);
            RuntimeDataService.RegisterOrUpdateFetchTask(task);

            foreach (var fetcherId in task.FetcherIds)
            {
                _fetcher2TaskMap.TryAdd(fetcherId, task);
                DispatchFetcher(fetcherId).Wait();
            }

            var receipt = new FetchReceipt()
            {
                Id = task.Id,
                FetcherIds = task.FetcherIds
            };

            Log.Information("LocalFetchFactory start Fetch with order {@order},receipt {@receipt}", order, receipt);
            return receipt;
        }

        public override void SpeedUpFetch(FetchReceipt receipt, int fetchers2Hire)
        {
            for (int i = 0; i < fetchers2Hire; i++)
            {
                var fetcherId = IdHelper.FetcherIdGenerator.GenerateId();

                if (_fetcher2TaskMap.TryGetValue(receipt.FetcherIds.First(), out FetchTask task))
                {
                    task.FetcherIds.Add(fetcherId);
                }
                receipt.FetcherIds.Add(fetcherId);
            }
            Log.Information("task {receiptId} on LocalFetchFactory is speed up,current ids is {ids}", receipt.Id, receipt.FetcherIds);

        }

        public override void SpeedDownFetch(FetchReceipt receipt, int fetchers2Dismiss)
        {
            if (receipt.FetcherIds.Count <= fetchers2Dismiss)
            {
                throw new ArgumentException("一次入库任务至少要保留一个fetcher");
            }
            var fetcherIds = receipt.FetcherIds.OrderBy(f => Guid.NewGuid()).Take(fetchers2Dismiss).ToList();
            foreach (var fetcherId in fetcherIds)
            {
                DismissFetcher(fetcherId);
                receipt.FetcherIds.Remove(fetcherId);
            }
            Log.Information("task {receiptId} on LocalFetchFactory is speed down,current ids is {ids}", receipt.Id, receipt.FetcherIds);
        }

        public override void StopFetch(FetchReceipt receipt)
        {
            Log.Information("Fetch {@receipt} is stopped", receipt);
            StopFetchers(receipt.FetcherIds);
        }

        private void StopFetchers(IList<long> fetcherIds)
        {
            foreach (var fetcherId in fetcherIds)
                if (_fetchers.TryGetValue(fetcherId, out var fetcher))
                    fetcher.Stop();
        }

        public void StartFetchers(IList<long> fetcherIds)
        {
            foreach (var fetcherId in fetcherIds)
            {
                if (_fetchers.TryGetValue(fetcherId, out var fetcher))
                {
                    if (fetcher.Status != FetcherStatus.Finished)
                    {
                        fetcher.WorkUntilSignal();
                    }
                }
            }

        }

        public override void ContinueFetch(FetchReceipt receipt)
        {
            Log.Information("Fetch {@receipt} is continued", receipt);
            StartFetchers(receipt.FetcherIds);
        }

        public async void FetchFinished(long taskId)
        {
            try
            {
                //爬行结束时Fetcher会自行关闭，不需要另外做关闭操作
                //标记PagePool状态为只读
                var pool = await GetPagePool(taskId);
                pool.ReadOnly = true;
                //释放Scheduler
                var scheduler = await _schedulers.GetComponentAsync(taskId,CurrentFactoryConfig.SchedulerType);
                scheduler?.Dispose();
                RuntimeDataService.UpdateFetchTaskStatus(taskId, FetchTaskStatus.Solved);
                Log.Information("fetch task {id} is finished", taskId);
            }
            catch (Exception e)
            {
                Log.Warning("fetch {} finished un clearly,{exception}", taskId, e);
            }

        }

        #endregion

        #region Fethcer调度

        /// <summary>
        /// 以Fetcher为单位颗粒度调度，因为会缓存所以没必要多个集中分配
        /// </summary>
        /// <param name="fetcherId"></param>
        public async Task DispatchFetcher(long fetcherId)
        {
            try
            {
                if (_fetchers.ContainsKey(fetcherId)) return;

                //await不阻塞线程，所以DispatchFetcher不影响主线程的退出
                //但是系统期望的效果是等待分配的fetcher完成工作，所以改为获得需要wait的Result
                var task = GetFetchTaskByFetcherIdCached(fetcherId);
                if (task == null)
                {
                    throw new ArgumentException($"can't find task info by fetcher id={fetcherId}'");
                }

                var taskId = task.Id;
                var fetcher = GetInstance<AbstractFetcher>(task.FetchOrder.FetcherTypeName);

                 fetcher.Initialize(fetcherId, taskId, task.FetchOrder,
                    await _schedulers.GetComponentAsync(taskId,CurrentFactoryConfig.SchedulerType), await GetPagePool(taskId),await GetProxyPool(taskId)
                    , GetLinkExtractor(taskId, task.FetchOrder), 
                    await _detectors.GetComponentAsync(taskId, typeof(AntiSpiderDetector).FullName)).Wait();
                fetcher.WorkUntilSignal();
                _fetchers.TryAdd(fetcherId, fetcher);

                UnitSchedules.DispatchFetcher(fetcherId, task.FetchOrder.FetcherTypeName);
                UpdateScheduleInfo();
                Log.Debug("fetcher {id} is dispatched to factory {factory}", fetcherId, Id);
            }
            catch (Exception e)
            {
                Log.Warning("fetcher {id} dispatch failed on {endpoint},because of {exception}", fetcherId, UnitSchedules.Endpoint, e);
            }
        }

        /// <summary>
        /// 辞退fetcher，与派遣(dispatch)fetcher对应
        /// </summary>
        /// <param name="fetcherId"></param>
        public void DismissFetcher(long fetcherId)
        {
            try
            {
                if (_fetchers.TryRemove(fetcherId, out var fetcher))
                {
                    fetcher.Stop();
                    UnitSchedules.DismissFetcher(fetcherId, fetcher.GetType().FullName);
                }
                UpdateScheduleInfo();
            }
            catch (Exception e)
            {
                Log.Warning("fetcher {id} dismiss failed on {endpoint},because of {exception}", fetcherId, UnitSchedules.Endpoint, e);
            }
        }

        #endregion

        #region 进度监控

        public override void Watch(FetchReceipt receipt, Action<FetchTaskProgress> progressAction)
        {
            var scheduler = _schedulers.GetComponentAsync(receipt.Id,CurrentFactoryConfig.SchedulerType).Result;
            if (scheduler == null) return;
            new Thread(() =>
            {
                FetchTaskProgress progress;
                do
                {
                    progress = scheduler.GetFetchProgress();
                    progressAction?.Invoke(progress);
                    Thread.Sleep(5 * 1000);
                } while (progress.Status != FetchTaskStatus.Canceled && progress.Status != FetchTaskStatus.Solved);

            }).Start();
        }

        #endregion

        #region 数据加工
        /// <summary>
        ///     数据加工没有分散到不同的机子上
        /// </summary>
        /// <param name="processOrder"></param>
        public override ProcessReceipt Process(ProcessOrder processOrder)
        {
            var processReceipt = new ProcessReceipt()
            {
                Id = IdHelper.ProcessTaskIdGenerator.GenerateId()
            };
            Log.Information("receipt {receipt} is processing by {processor}", processOrder.FetchReceiptId, processOrder.ProcessorType);
            var pool = GetPagePool(processOrder.FetchReceiptId).Result;
            var coordinator = GetProcessCoordinator(processReceipt.Id).Result;
            var consumer = new PagePoolConsumer(pool, coordinator);
            consumer.AddPageProcessor(GetInstance<IPageProcessor>(processOrder.ProcessorType));
            consumer.StartConsume(-1, processOrder.ProcessorCount);
            return processReceipt;
        }

        public override PagePoolConsumer Consume(FetchReceipt receipt)
        {
            Log.Information("process the fetched pages of fetch {id}", receipt.Id);
            var pool = GetPagePool(receipt.Id).Result;
            var coordinator = GetProcessCoordinator(IdHelper.ProcessTaskIdGenerator.GenerateId()).Result;
            var consumer = new PagePoolConsumer(pool, coordinator);

            return consumer;
        }


        #endregion

        #region 工具方法

        /// <summary>
        /// 持久化workshop的调度信息。注意，这里的ScheduleData信息本身已经被外部改变才会登记
        /// </summary>
        public void UpdateScheduleInfo()
        {
            RuntimeDataService.UpdateWorkshopScheduleData(UnitSchedules);
        }

        private FetchTask GetFetchTaskByFetcherIdCached(long fetcherId)
        {
            ;
            if (_fetcher2TaskMap.TryGetValue(fetcherId, out FetchTask task))
            {
                return task;
            }

            return RuntimeDataService.GetFetchTaskByFetcherId(fetcherId);
        }

        #endregion
    }
}
